package com.lin.Rpc;

import com.lin.util.RabbitUtil;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

import java.io.Closeable;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

public class Client implements Closeable {
    private String requestQueueName = "fibo_rpc_queue";
    private Connection connection;
    private Channel channel;

    public Client() throws IOException {
        connection = RabbitUtil.getConnection();
        channel = connection.createChannel();
    }

    public String Call(String message) throws Exception {
        final String uuid = UUID.randomUUID().toString();

//      使用默认队列来当作接受答复的队列--这里可以换成全局共享的队列--但是就需要特殊处理了-.- 比如消息需要可重复消费
        String replyQueueName = channel.queueDeclare().getQueue();
        System.out.println(replyQueueName);
//        AMQP协议
        AMQP.BasicProperties props = new AMQP.BasicProperties
                .Builder()
                .correlationId(uuid)
                .replyTo(replyQueueName)
                .build();

        channel.basicPublish("",requestQueueName,props,message.getBytes(StandardCharsets.UTF_8));

        final BlockingQueue<String> response =  new ArrayBlockingQueue<String>(1);

        String ctag = channel.basicConsume(replyQueueName,true,(consumerTag,delivery)->{
            if(delivery.getProperties().getCorrelationId().equals(uuid)) {
                response.offer(new String(delivery.getBody(),StandardCharsets.UTF_8));
            }
        },consumerTag->{});
        String result = response.take();
//      取消订阅这个队列
        channel.basicCancel(ctag);
        return result;
    }

    @Override
    public void close() throws IOException {
        connection.close();
    }
}
